package com.nuonuo.middleware.config;

import com.google.common.collect.Lists;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;

/**
 * @author duan
 * @version 1.0
 * @date 2023/12/28 15:19
 */
public class MessageConsumer {
    private final KafkaConsumer<String, String> consumer;

    private volatile Iterator<ConsumerRecord<String, String>> iterator;

    public MessageConsumer() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "192.168.210.200:9097,192.168.210.201:9097,192.168.210.202:9097");
        properties.put("group.id", "duan-fpmp");
        properties.put("client.id", "duan-fpmp-test");
        properties.put("max.poll.records", "10");
        properties.put("session.timeout.ms", "30000");
        properties.put("enable.auto.commit", "false");
        properties.put("key.deserializer", StringDeserializer.class.getName());
        properties.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<>(properties);
    }


    public void start() {
        //处理业务
        new Thread(new ConsumerThread(),"Thread-duan-fpmp-test").start();
    }

    private class ConsumerThread extends Thread {
        @Override
        public void run() {
            consumer.subscribe(Collections.singleton("duan-fpmp-topic"));
            ConsumerRecords<String, String> records;

            while (true) {
                try {
                    records = consumer.poll(1000);
                    if (records.isEmpty()) {
                        continue;
                    }
                    // 提前提交，防止消息处理时间超过session超时时间，导致提交失败。
                    // 消息完整性由业务系统提供保证。
                    consumer.commitSync();
                } catch (Throwable throwable) {
                    throwable.printStackTrace();
                    continue;
                }

                //数组迭代会报错，换成用List去装
                List<ConsumerRecord<String, String>> iteratorList = Lists.newArrayList();
                records.forEach(iteratorList::add);
                iterator = iteratorList.iterator();
                while(iterator.hasNext()){
                    ConsumerRecord<String, String> record = iterator.next();
                    try {
//                        callback.onMessage(record.key(), record.value(),record.topic(), record.partition());
//                        ConsumerHandler.addQueue(record.value());
                        ConsumerHandler.executorService.execute(new ConsumerHandler.FpmpConsumerHandler(record.value()));
                        iterator.remove();
                    } catch (Throwable throwable) {
                        throwable.printStackTrace();
                    }
                }
            }
        }
    }
}
